package job2;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.List;

// 实现一个协处理器,并上传到hdfs上，并挂载
// 删除好友操作实现（好友关系双向，一方删除好友，另一方也会被迫删除好友）
public class DeleteRelationsProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
//        获取用户关系表
        HTable relationship = (HTable) e.getEnvironment().getTable(TableName.valueOf("relationship"));
//          获取本次删除命令包含friends 列族的单元格
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));
//       确保不会索引越界
        if (!CollectionUtils.isEmpty(cells)) {
            // 获取第一个单元格
            Cell cell = cells.get(0);
            //获取单元格的uid
            byte[] uid = CellUtil.cloneRow(cell);
            //获取单元格的fid
            byte[] fid = CellUtil.cloneQualifier(cell);
//            将fid和uid 交换顺序
            Delete otherUserDelete = new Delete(fid);
            otherUserDelete.addColumns(Bytes.toBytes("friends"), uid);
//            在fid 的列表，也删除uid 好友关系
            relationship.delete(otherUserDelete);
        }
        // 关闭 table 对象
        relationship.close();
    }
}
